package com.zk.storm;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.io.IOException;
import java.util.Map;

public class HBaseWriteBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Connection connection;

    // 聚合类型对应的表名
    private static final String COUNTRY_TABLE = "country"; //国家或地区
    private static final String PLATFORM_TABLE = "platform";//操作系统
    private static final String DEVICE_TABLE = "device";//设备
    private static final String AGENT_TABLE = "agent";//浏览器
    private static final byte[] CF_BYTES = Bytes.toBytes("cf");

    // 列名常量
    private static final String SUM_HITS = "sum_hits";
    private static final String AVG_HITS = "avg_hits";
    private static final String MAX_HITS = "max_hits";

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        try {
            // 初始化HBase配置和连接
            org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
            hbaseConf.set("hbase.zookeeper.quorum", "localhost:2183");
            connection = ConnectionFactory.createConnection(hbaseConf);
        } catch (IOException e) {
            throw new RuntimeException("HBase连接初始化失败: " + e.getMessage());
        }
    }

    @Override
    public void execute(Tuple input) {
        try {
            // 提取公共字段
            String aggKey = input.getStringByField("agg_key");
            String sourceComponent = input.getSourceComponent();//获取来源组件名称
            // 根据来源组件确定目标表和列
            TableName tableName = getTargetTable(sourceComponent);//调用存储目标表映射
            String column = getTargetColumn(sourceComponent); //调用数据类型存储映射

            // 获取表连接
            try (Table table = connection.getTable(tableName)) {
                Put put = new Put(Bytes.toBytes(aggKey));

                // 根据列类型添加值
                if (SUM_HITS.equals(column)) {
                    long sumHits = input.getLongByField(SUM_HITS);
                    put.addColumn(CF_BYTES, Bytes.toBytes(column), Bytes.toBytes(sumHits));
                }
                else if (AVG_HITS.equals(column)) {
                    double avgHits = input.getDoubleByField(AVG_HITS);
                    put.addColumn(CF_BYTES, Bytes.toBytes(column), Bytes.toBytes(avgHits));
                }
                else if (MAX_HITS.equals(column)) {
                    int maxHits = input.getIntegerByField(MAX_HITS);
                    put.addColumn(CF_BYTES, Bytes.toBytes(column), Bytes.toBytes(maxHits));
                }
                // 执行写入
                table.put(put);
            }

            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
            throw new RuntimeException("HBase写入失败: " + e.getMessage());
        }
    }

    // 映射转化
    private TableName getTargetTable(String sourceComponent) {
        switch (sourceComponent) {
            case "country-sum-bolt":
                return TableName.valueOf(COUNTRY_TABLE);
            case "platform-avg-bolt":
                return TableName.valueOf(PLATFORM_TABLE);
            case "device-max-bolt":
                return TableName.valueOf(DEVICE_TABLE);
            case "agent-sum-bolt":
                return TableName.valueOf(AGENT_TABLE);
            default:
                throw new IllegalArgumentException("未知来源组件: " + sourceComponent);
        }
    }

    // 根据来源组件确定目标列
    private String getTargetColumn(String sourceComponent) {
        switch (sourceComponent) {
            case "country-sum-bolt":
            case "agent-sum-bolt":
                return SUM_HITS;
            case "platform-avg-bolt":
                return AVG_HITS;
            case "device-max-bolt":
                return MAX_HITS;
            default:
                throw new IllegalArgumentException("未知来源组件: " + sourceComponent);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }

    @Override
    public void cleanup() {
        // 关闭连接
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                System.err.println("HBase连接关闭异常: " + e.getMessage());
            }
        }
    }
}